package cn.com.demo.kafka.producer;

import cn.com.demo.kafka.Config;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.*;

import static org.apache.kafka.clients.producer.ProducerConfig.*;
import static org.apache.kafka.streams.StreamsConfig.BOOTSTRAP_SERVERS_CONFIG;
/**
 * kafka 发数据
 */
public class ProducerAutoDemo {

    public static void main(String args[]) {
        Properties conf = new Properties();
        conf.setProperty(BOOTSTRAP_SERVERS_CONFIG, Config.BOOTSTRAP_SERVERS_CONFIG);
        conf.put(ACKS_CONFIG, "all");
        conf.put(KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        conf.put(VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String, String> producer = new KafkaProducer<>(conf);

        for (;;) {
            int i = new Random().nextInt(100);
            String val = UUID.randomUUID().toString();

            ProducerRecord<String, String> record = new ProducerRecord<>(Config.TOPIC, i+"--"+val, val);
            try {
                System.out.println("kafka send ["+i+"]:"+val);
                // 简单的直接send，在消费的时候，是消费不到的
                //producer.send(record);
                producer.send(record).get();
                Thread.sleep(Config.PRODUCER_SEND_INTERVAL);
            } catch (Exception e){
                System.out.println(e.toString());
            }
        }

    }
}